package com.hoveagle.kafka.consumer;

import com.hoveagle.kafka.exceptions.ReadFileException;
import com.hoveagle.kafka.handler.MessageHandler;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * @author hoveagle
 */
public class BusinessConsumer<K,V> implements MessageConsumer{
    private static final Logger logger = LoggerFactory.getLogger(BusinessConsumer.class);
    private KafkaConsumer<K,V> kafkaConsumer;
    private MessageHandler<K,V> messageHandler;
    private KafkaConfig kafkaConfig;
    private AtomicBoolean running = new AtomicBoolean(false);

    public BusinessConsumer(KafkaConfig consumerConfig, MessageHandler messageHandler){
        this.kafkaConfig = consumerConfig;
        this.messageHandler = messageHandler;
    }

    @Override
    public void start() {
        initKafkaConsumer();
       this.running.set(true);
    }

    @Override
    public void stop() {
       this.running.set(false);
       shutDown();
    }

    @Override
    public void run() {
       while(running.get()){
           doWork();
       }
        shutDown();
    }


    private void initKafkaConsumer(){
        boolean reconnect = true;
        while(reconnect){
          try{
              this.kafkaConsumer = new KafkaConsumer<>(this.kafkaConfig.getKafkaConfig());
              this.kafkaConsumer.subscribe(Arrays.asList(this.kafkaConfig.getTopics()));
              reconnect = false;
          } catch (ReadFileException e) {
              logger.warn("Failed to init kafka kafka",e);
              reconnectionPooling();
          }
        }
    }

    private void doWork(){
        try{
            final long timeout = 1000;
            ConsumerRecords<K, V> records = this.kafkaConsumer.poll(timeout);
            if(records != null && !records.isEmpty()){
                for(ConsumerRecord<K,V> consumerRecord: records){
                    this.messageHandler.onMessage(consumerRecord);
                }
                this.kafkaConsumer.commitSync();
            }
        }catch (Exception e){
            logger.warn("Failed to poll kafka records",e);
            reconnectionPooling();
        }
    }

    private void reconnectionPooling() {
        try{
            TimeUnit.MINUTES.sleep(1);
        } catch (InterruptedException e) {
            logger.warn("Failed to reconnect",e);
        }
    }

    private void shutDown(){
        if(this.kafkaConsumer != null){
            this.kafkaConsumer.unsubscribe();
            this.kafkaConsumer.wakeup();
            this.kafkaConsumer.close();
        }
    }
}
